package com.bungarus.busi;

import com.bungarus.Transfer;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.BungarusClientException;
import com.bungarus.model.BungarusServer;
import com.bungarus.model.MessageBuilder;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by tang on 2018/11/30.
 */
public class TransferRemedy extends Remedy {
    private List<BaseClient> innerClients ;
    private static boolean started = false;

    public synchronized static void start(BungarusServer server0) throws Exception {
        if(!started) {
            started = true;
            new TransferRemedy().run(server0);
        }
    }

    @Override
    protected void run(BungarusServer server) throws Exception {
        innerClients = ((Transfer)server).getTransferInnerClients();
        executorGroup.scheduleAtFixedRate(() -> {
            DischargeQueue queue = Discharger.getQueueInstance();
            TransferRemedy.this.retry0(queue);

            logger.info("Remedy is running...");
        }
                ,DELAY_TIME ,PERIOD , TimeUnit.SECONDS);
        logger.info("Remedy is running...");
    }

    private void retry0(DischargeQueue queue) {
        logger.debug("Size of discharge queue: " + queue.size());
        if(queue.size() <= 0 || innerClients.isEmpty()){
            return;
        }
        ConcurrentHashMap.KeySetView<Long, MessageProtobuf.Message> keys = queue.keys();
        for(Long l: keys){
            MessageProtobuf.Message noAckMsg = queue.getFromQueue(l);
            if((System.currentTimeMillis() - noAckMsg.getHeader().getSentTime()) >= ACK_TIMEOUT) {
                MessageProtobuf.Extension extension = noAckMsg.getHeader().getExtensions(0);
                if(null != extension && extension.getKey().equalsIgnoreCase(MessageBuilder.TAG)) {
                    String tag = extension.getValue();
                    queue.removeFromQueue(l);
                    innerClients.forEach(baseClient -> {
                        try {
                            baseClient.sendMessage(MessageBuilder.buildRemedyMessage(noAckMsg), tag);
                        }
                        catch (BungarusClientException e) {
                            e.printStackTrace();
                        }
                    });
                }
            }
        }
    }
}
